//package com.example.kafka.flink;
//
//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.api.common.state.ValueState;
//import org.apache.flink.api.common.state.ValueStateDescriptor;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.streaming.api.TimeCharacteristic;
//import org.apache.flink.streaming.api.datastream.DataStreamSource;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
//import org.apache.flink.streaming.api.functions.source.SourceFunction;
//import org.apache.flink.util.Collector;
//
//import java.time.Duration;
//
//public class eventTimeWindow {
//
//    public static void eventTimeWindow() throws Exception {
//        long ct=System.currentTimeMillis();
//        StreamExecutionEnvironment e = StreamExecutionEnvironment.getExecutionEnvironment();
//        e.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//        e.getConfig().setAutoWatermarkInterval(1000);
//        DataStreamSource<Long> source = e
//                .addSource(new SourceFunction<Long>() {
//                    private volatile boolean stop = false;
//                    @Override
//                    public void run(SourceContext<Long> ctx) throws Exception {
//                        for(int j=0;j<10;j++) {
//                            ctx.collectWithTimestamp(
//                                    (long) j,
//                                    ct);
//                            Thread.sleep(500);
//                        }
//                        for(int j=0;j<10;j++) {
//                            ctx.collectWithTimestamp(
//                                    (long) j,
//                                    System.currentTimeMillis());
//                            Thread.sleep(500);
//                        }
//                    }
//                    @Override
//                    public void cancel() {
//                        stop = true;
//                    }
//                }).setParallelism(1);
//        source.assignTimestampsAndWatermarks(WatermarkStrategy
//                        .<Long>forBoundedOutOfOrderness(Duration.ofSeconds(1)))
//                .keyBy(v->v/1000).process(new KeyedProcessFunction<Long, Long, Long>() {
//                    private ValueState<Integer> itemState;
//                    @Override
//                    public void open(Configuration parameters) throws Exception {
//                        super.open(parameters);
//                        ValueStateDescriptor<Integer> itemsStateDesc = new ValueStateDescriptor<>(
//                                "itemState-state",
//                                Integer.class);
//                        itemState = getRuntimeContext().getState(itemsStateDesc);
//                    }
//                    @Override
//                    public void processElement(Long value, Context ctx, Collector<Long> out) throws Exception {
//                        int val=(itemState.value()==null)?0:itemState.value();
//                        itemState.update(val+1);
//                        System.out.println(ctx.timerService().currentWatermark()+","+ctx.timestamp());
//                        ctx.timerService().registerEventTimeTimer(ct);
//                    }
//                    @Override
//                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Long> out) throws Exception {
//                        super.onTimer(timestamp, ctx, out);
//                        System.out.println(itemState.value()+"---"+timestamp+"——"+ctx.getCurrentKey());
//                    }
//                    @Override
//                    public void close() throws Exception {
//                        super.close();
//                    }
//                }).setParallelism(1);
//        e.execute();
//    }
//}
